package com.ctbri.manage.bydeequ.example
import com.amazon.deequ.VerificationSuite
import com.amazon.deequ.checks.{Check, CheckLevel, CheckStatus}
import com.amazon.deequ.constraints.ConstraintStatus
import org.apache.spark.sql.{Dataset, Row, SparkSession}

import java.util.{List, Map}
/**
 * @author songyunlong
 * @createTime 2023/5/17 22:10
 * @description
 */
object TestExampleWithRealData {
    def calc(sc: SparkSession, dataFrame: Dataset[Row], needShow: Boolean = true, rowNumber: Int = 20,
             colWithOp: Map[String, String] = null, totalTableOp: List[String] = null): Unit = {
        if (needShow) {
            try {
                if (rowNumber <= 0) throw new Exception("rowNumber less than zero!")
                dataFrame.show(rowNumber)
            } catch {
                case ex: Exception => {
                    println("rowNumber need greater than zero!")
                }
            } finally {
                val verificationResult = VerificationSuite()
                    .onData(dataFrame)
                    .addCheck(
                        Check(CheckLevel.Error, "unit testing my data")
                            .hasSize(_ == 30) // we expect 5 rows
                            .isComplete("ppm_code") // should never be NULL
                            .isUnique("id") // should not contain duplicates
                            .isComplete("parent_id") // should never be NULL
                            // should only contain the values "high" and "low"
                            .isContainedIn("is_valid", Array("100383", "100387"))
                            .isNonNegative("area_level_id") // should not contain negative values
                            // at least half of the descriptions should contain a url
//                            .containsURL("description", _ >= 0.5)
//                            // half of the items should have less than 10 views
//                            .hasApproxQuantile("numViews", 0.5, _ <= 10)
                    )
                    .run()
                if (verificationResult.status == CheckStatus.Success) {
                    println("The data passed the test, everything is fine!")
                } else {
                    println("We found errors in the data:\n")

                    val resultsForAllConstraints = verificationResult.checkResults
                        .flatMap { case (_, checkResult) => checkResult.constraintResults }

                    resultsForAllConstraints
                        .filter {
                            _.status != ConstraintStatus.Success
                        }
                        .foreach { result => println(s"${result.constraint}: ${result.message.get}") }
                }
                sc.stop()
            }

        }
    }
    def main(args: Array[String]): Unit = {
        println("Hello Scala!")
    }
}
